package com.bw.app.dwd1;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.utils.DateFormatUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.Random;

/**
 * @projectName: Flinksxe
 * @fileName: BaseLogApp
 * @packageName: com.bw.app.dwd
 * @date: 2025/9/11 10:31
 * @author: ymw
 */


/**
 * 首先要读取日志数据
 *
 */
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        //状态后端
//        env.setStateBackend(new HashMapStateBackend());
//        //检查点
//        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        String topic = "topic_log";
        String groupId = "base_log_consumer";
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirstream") {};
        SingleOutputStreamOperator<JSONObject> clearStream = source.process(
                new ProcessFunction<String, JSONObject>() {
                    @Override
                    public void processElement(String value, ProcessFunction<String, JSONObject>.Context ctx,
                                               Collector<JSONObject> out) throws Exception {
                        try {
                            if (JSONObject.isValid(value)) {
                                JSONObject jsonObject = JSONObject.parseObject(value);
                                out.collect(jsonObject);
                            }
                        } catch (Exception e) {
                            ctx.output(dirtyStreamTag, value);
                            System.out.println(value);
                        }
                    }
                }
        );
//        DataStream<String> dirtyStream = clearStream.getSideOutput(dirtyStreamTag);
//        String dirTopic = "dirty_data";
//        dirtyStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(dirTopic));
//        dirtyStream.print("脏数据：");
//        clearStream.print("正常数据：");
        KeyedStream<JSONObject, String> keyedStream = clearStream.keyBy(r -> r.getJSONObject("common").getString("mid"));
        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<String> firstViewDtState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                firstViewDtState = getRuntimeContext().getState(new
                        ValueStateDescriptor<String>("firstViewDt", String.class));
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject,
                    JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                String isNew = value.getJSONObject("common").getString("is_new");
                //状态时间
                String firstViewDt = firstViewDtState.value();
                //获取当前数据的时间
                Long ts = value.getLong("ts");
                //把时间戳转换为年月日
                String dt = DateFormatUtil.toDate(ts);
                if ("1".equals(isNew)) {
                    if (firstViewDt == null) {
                        firstViewDtState.update(dt);
                    } else {
                        if (!firstViewDt.equals(dt)) {
                            isNew = "0";
                            value.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    if (firstViewDt == null) {
                        String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                        firstViewDtState.update(yesterday);
                    }
                }
                out.collect(value);
            }
        });
        Random random = new Random();
        OutputTag<String> startTag = new OutputTag<String>("startTag") {};
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {};
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {};
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {};
        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx,
                                       Collector<String> out) throws Exception {
                //收集错误数据
                JSONObject error = value.getJSONObject("err");
                if (error != null) {
                    ctx.output(errorTag, value.toJSONString());
                }
                value.remove("err");
                //收集启动数据
                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    ctx.output(startTag, value.toJSONString());
                } else {
                    JSONObject page = value.getJSONObject("page");
                    JSONObject common = value.getJSONObject("common");
                    if (page != null) {
                        page.put("shop_id", "shop_" + random.nextInt(1000));
                    }
                    Long ts = value.getLong("ts");
                    //收集曝光数据
                    JSONArray displays = value.getJSONArray("displays");
                    if (displays != null) {
                        for (int i = 0; i < displays.size(); i++) {
                            JSONObject display = displays.getJSONObject(i);
                            JSONObject displayObj = new JSONObject();
                            displayObj.put("display", display);
                            displayObj.put("common", common);
                            displayObj.put("page", page);
                            displayObj.put("ts", ts);
                            ctx.output(displayTag, displayObj.toJSONString());
                        }
                    }
                    //收集动作数据
                    JSONArray actions = value.getJSONArray("actions");
                    if (actions != null) {
                        for (int i = 0; i < actions.size(); i++) {
                            JSONObject action = actions.getJSONObject(i);
                            JSONObject actionObj = new JSONObject();
                            actionObj.put("action", action);
                            actionObj.put("common", common);
                            actionObj.put("page", page);
                            actionObj.put("ts", ts);
                            ctx.output(actionTag, actionObj.toJSONString());
                        }
                    }
                    //收集页面数据
                    value.remove("displays");
                    value.remove("actions");
                    out.collect(value.toJSONString());
                }
            }
        });
        //打印主流和各侧输出流查看分流效果
        separatedStream.print("page====>");
        separatedStream.getSideOutput(startTag).print("start====>");
        separatedStream.getSideOutput(displayTag).print("display====>");
        separatedStream.getSideOutput(actionTag).print("action====>");
        separatedStream.getSideOutput(errorTag).print("error====>");
        //提取各侧输出流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);

        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_page_log"));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_start_log"));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_display_log"));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_action_log"));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer("dwd_traffic_error_log"));
        env.execute("BaseLogApp");
    }
}
